package demo.kafka.web;
import demo.kafka.config.ConsumerCfg;
import demo.kafka.config.ProducerCfg;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.web.bind.annotation.*;

import java.time.LocalDateTime;
import java.util.UUID;

@RestController
@RequestMapping("/api/messages")
public class MessageController {

    private final StreamBridge streamBridge;
    private final ConsumerCfg messageConsumer;

    public MessageController(StreamBridge streamBridge, ConsumerCfg messageConsumer) {
        this.streamBridge = streamBridge;
        this.messageConsumer = messageConsumer;
    }

    /**
     * 手动发送消息的接口
     */
    @PostMapping("/send")
    public String sendMessage(@RequestBody String content) {
        ProducerCfg.MessageEvent event = new ProducerCfg.MessageEvent(
                UUID.randomUUID().toString(),
                content,
                "USER",
                LocalDateTime.now(),
                -1  // 手动发送的消息序列号为-1
        );

        // 使用StreamBridge手动发送消息
        boolean sent = streamBridge.send("topic_stan", event);

        if (sent) {
            return "Message sent successfully: " + event.getMessageId();
        } else {
            return "Failed to send message";
        }
    }

    /**
     * 获取消费统计
     */
    @GetMapping("/stats")
    public String getStats() {
        return "Total messages consumed: " + messageConsumer.getMessageCount();
    }

    /**
     * 健康检查端点
     */
    @GetMapping("/health")
    public String health() {
        return "Kafka Stream Demo Application is running!";
    }
}